package MapReduce;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class Mapper02 extends
		Mapper<LongWritable, Text, Text, IntWritable> {

	private final static IntWritable one = new IntWritable(1);
	private Text word = new Text();

	public void map(LongWritable key, Text value, Context context)
			throws IOException, InterruptedException {
		SimpleDateFormat hdsdf = new SimpleDateFormat("yyyy/MM");
		Date lastmonth = null;
		Calendar cal = Calendar.getInstance();
		
		String[] colums = value.toString().split(",");
		String[] date = colums[0].toString().split("/");
		
		String[] chk = new String[3];
		String chkyear = "";
		String chkmonth = "";
		
		cal.set(Calendar.DATE, cal.get(Calendar.MONTH) - 1);
		lastmonth = cal.getTime();
		chk = hdsdf.format(lastmonth).toString().split("/");
		chkyear = chk[0];
		chkmonth = chk[1];
		
		if(date[1].equals(chkmonth)){
			word.set(colums[2]);
			context.write(word, one);
		}
	}
}